package preprocess;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import trouger.dmt.OutputRecordFilter;
import trouger.dmt.SelfAggregationRunner;
import trouger.dmt.SourceRecordFilter;
import utils.Utils;

public class UserFilter implements Tool{

	public static String getFilteredUserPath(String date) {
		return Utils.rootDir + date + "/UserFiltered/";
	}
	
	public static class SourceFilter implements SourceRecordFilter {
		@Override
		public boolean isValidRecord(int sourceIndex, String[] segment) {
			return segment.length == 28 && segment[1].equals("c1") && Utils.isValid(segment[4]);
		}
	}
	
	public static class OutputFilter implements OutputRecordFilter{
		@Override
		public boolean filterRecord(Object[] record) {
			int clickCount  = Integer.parseInt(record[2].toString());
			return clickCount < 200 && clickCount >= 5; 
		}
	}
	
	@Override 
	public Configuration getConf() {
		return null;
	}
	
	@Override
	public void setConf(Configuration conf) {
		
	}
	/*
	 * (non-Javadoc)
	 * UserFiltered Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/UserFiltered/
	 * 1.数字id
	 * 2.点击数
	 */
	@Override
	public int run(String[] args) throws Exception {
		String clickLogFile = "/group/taobao/taobao/dw/fact/" + args[0] + "/NewSearchLog/";
		String output = getFilteredUserPath(args[0]);
		System.out.println("start job UserFilter...");
		System.out.println("click: " + clickLogFile);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		SelfAggregationRunner r = new SelfAggregationRunner();
		r.addSourceTable(clickLogFile, "UTF-8", "\t");		
		r.addKeyGroup("1.4");
		r.setResultFieldsEx("1.4, Count(1.4)");		
		r.setSourceRecordFilterClass(SourceFilter.class);
		r.setOutputRecordFilterClass(OutputFilter.class);
		r.setResultTable(output);
		r.run("UserFilterJob");
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new UserFilter(), args);
		System.exit(res);
	}
}
